package com.aotain.hbase;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;


public class HBaseDao implements Serializable{

//	private static Connection connection ;
	private static HConnection connection ;
//	private  HConnection connection ;
	private static String _zooServer = "192.168.5.95";

	static HBaseDao instance ;
	
	public static HBaseDao getInstance(String zkserver) {
		if(instance == null) {
			synchronized(HBaseDao.class) {
				instance = new HBaseDao(zkserver);
				InitConnection();
			}
		}
//		InitConnection();
		return instance;
	}

	public HBaseDao()
	{
		//		Configuration conf = new Configuration();
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("hbase.zookeeper.quorum", _zooServer);
//		try {
//						connection = HConnectionManager.createConnection(conf) ;
//			connection =  ConnectionFactory.createConnection(conf) ;
//		} catch (IOException e) {
//			e.printStackTrace();
//		}
	}

	public HBaseDao(String _zooServer) {
		this._zooServer = _zooServer;
//		Configuration conf = HBaseConfiguration.create();
//		conf.set("hbase.zookeeper.property.clientPort", "2181");
//		conf.set("hbase.zookeeper.quorum", _zooServer);
//		try {
//			connection = HConnectionManager.createConnection(conf) ;
//		} catch (IOException e) {
//			e.printStackTrace();
//		}
		
	}

	public  void closeConn() {
		if (connection != null) {
			try {
				connection.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
	public static  void InitConnection() {
		Configuration hConfig = HBaseConfiguration.create();
		hConfig.set("hbase.zookeeper.quorum", _zooServer);
		hConfig.set("hbase.zookeeper.property.clientPort","2181");

		try {
//			connection = ConnectionFactory.createConnection(hConfig);
			connection = HConnectionManager.createConnection(hConfig) ;
		} catch (IOException e) {
			
			throw new RuntimeException(e);
		}
	}


	/** 
	 * @Title: save 
	 * @Description: 根据表名保存
	 * @param put
	 * @param tableName
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2015年8月21日 上午10:01:44
	 */ 
	public void save(Put put, String tableName) {
		// TODO Auto-generated method stub
		//		HTableInterface table = null;
//		InitConnection();
		HTable table = null;
	
		try { 
			//			table = connection.getTable(tableName) ;
			table = (HTable) connection.getTable(TableName.valueOf(tableName));
			table.setAutoFlushTo(false);
			table.setWriteBufferSize(8*1024*1024);
			table.put(put) ;

		} catch (Exception e) {
			e.printStackTrace() ;
		}finally{
			try {
				if(table != null)
				table.close() ;
				//				connection.close();
//				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/** 
	 * @Title: insert 
	 * @Description: TODO(插入指定rowkey 、列 到指定tableName的表中 ) 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param quailifer
	 * @param value
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2015年8月21日 上午10:02:16
	 */ 
	public void insert(String tableName, String rowKey, String family,
			String quailifer, String value) {
		// TODO Auto-generated method stub
		//		HTableInterface table = null;
		InitConnection();
		Table table = null;
		try {
			TableName TABLE_NAME = TableName.valueOf(tableName);
			table = connection.getTable(TABLE_NAME);
//						table = connection.getTable(tableName) ;
//			table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(rowKey.getBytes());
			put.addColumn(family.getBytes(), quailifer.getBytes(), value.getBytes()) ;
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		}finally
		{
			try {
				table.close() ;
				connection.close();
				//				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 
	 * @Title: insert 
	 * @Description: 多列添加
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param quailifer
	 * @param value
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2015年9月2日 下午4:29:30
	 */
	public void insert(String tableName,String rowKey,String family,String quailifer[],String value[])
	{
		//		HTableInterface table = null;
		InitConnection();
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName)) ;
			Put put = new Put(rowKey.getBytes());
			// 批量添加
			for (int i = 0; i < quailifer.length; i++) {
				String col = quailifer[i];
				String val = value[i];
				put.addColumn(family.getBytes(), col.getBytes(), val.getBytes());
			}
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		}finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/** 
	 * @Title: getRows 
	 * @Description: 得到指定列的行记录
	 * @param tableName
	 * @param rowKeyLike
	 * @param cols
	 * @return
	 * @return List<Result>    返回类型 
	 * @author 程彬
	 * @date 2015年9月7日 上午10:49:21
	 */ 
	public List<Result> getRows(String tableName, String rowKeyLike ,String cols[]) {
		// TODO Auto-generated method stub
		//		HTableInterface table = null;
		Table table = null;
		List<Result> list = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName)) ;
			PrefixFilter filter = new PrefixFilter(rowKeyLike.getBytes());
			Scan scan = new Scan();
			for (int i = 0; i < cols.length; i++) {
				scan.addColumn("cf".getBytes(), cols[i].getBytes()) ;
			}
			scan.setFilter(filter);
			ResultScanner scanner = table.getScanner(scan) ;
			list = new ArrayList<Result>() ;
			for (Result rs : scanner) {
				list.add(rs) ;
			}
		} catch (Exception e) {
			e.printStackTrace() ;
		}
		finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}

	public void incrementValue(String tableName,String rowKey,String cf ,String column,int amount) {
		Table table = null;
		Increment increment = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName)) ;
			increment = new Increment(Bytes.toBytes(rowKey));
			increment.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), amount);
			table.increment(increment);
		} catch(Exception e) {
			e.printStackTrace();
		} finally {
			try {

				table.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	/**
	 * 根据条件查询记录数
	 * @Title: QueryByKey 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tablename
	 * @param startRow
	 * @param endRow
	 * @param columns
	 * @return
	 * @return List<String>    返回类型 
	 * @author 程彬
	 * @date 2016年3月10日 下午4:36:46
	 */
	public int getAssignColumnNum(String tablename,String startRow,String endRow,String column) { 
		InitConnection();

		Table table = null;
		int rownum = 0;
		try { 

			TableName TABLE_NAME = TableName.valueOf(tablename);
			table = connection.getTable(TABLE_NAME);

			Scan scan = new Scan(); 
			//scan.setReversed(true);
			scan.setStartRow(startRow.getBytes());
			scan.setStopRow(endRow.getBytes());
			scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(column));

//			SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("cf"), Bytes.toBytes(column), CompareOp.EQUAL, null);
//			
//			scan.setFilter(filter);

			ResultScanner rs = table.getScanner(scan); 
			for (Result r : rs) {
				String rowkey = Bytes.toString(r.getRow());
				String row = "";
				row = rowkey;
				for(Cell c : r.listCells()){
					String cc = Bytes.toString(CellUtil.cloneQualifier(c));
//					if(column!=null && cc.equals(column)){
//						
						Long bb = Bytes.toLong(CellUtil.cloneValue(c));
						row = row + "_" + cc + "_" + bb;
						System.out.println("row:" + row);
						rownum ++;
//					}
				}
			}
		} catch (Exception e) { 
			e.printStackTrace(); 
		} 
		finally
		{
			try {
				table.close();
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		System.out.println("rownum =" + rownum);
		return rownum;
	} 

	public void incerment(String tableName,String rowKey,String cf ,String column ,int value) {
		InitConnection();

		Table table = null;
		
		TableName tablename = TableName.valueOf(tableName);
		try {
			table = connection.getTable(tablename);
			
			Increment increment = new Increment(Bytes.toBytes(rowKey));
			increment.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), value);
			
			table.increment(increment);
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally
		{
			try {
				table.close();
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	public List<String> getDataByCon(String tableName,String startRow,String endRow) {
		List<String> list = new ArrayList<String>();
		 InitConnection();
		
		 Table table = null;
       try { 

       	TableName TABLE_NAME = TableName.valueOf(tableName);
    		table = connection.getTable(TABLE_NAME);
   		
   		Scan scan = new Scan(); 
   		//scan.setReversed(true);
   		scan.setStartRow(startRow.getBytes());
   		scan.setStopRow(endRow.getBytes());
   		
   		
           ResultScanner rs = table.getScanner(scan); 
           for (Result r : rs) {
              String rowkey = Bytes.toString(r.getRow());
 			   String row = "";
 			   row = rowkey;
 			   
 			   list.add(row);
           }
       } catch (Exception e) { 
           e.printStackTrace(); 
       } 
       finally
       {
       	try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
       }
       return list;
	}
	
	public static void main(String[] args) {
		/**
		 * 本地运行时所需
		 */
		File workaround = new File("."); 
		System.getProperties().put("hadoop.home.dir", workaround.getAbsolutePath()); 
		new File("./bin").mkdirs();  

		try {
			new File("./bin/winutils.exe").createNewFile();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
//		HBaseDao dao = new HBaseDao("132.126.32.2,132.126.32.3,132.126.32.4"); 
		HBaseDao dao = HBaseDao.getInstance("132.126.32.2,132.126.32.3,132.126.32.4");
		Put put = new Put("cheng".getBytes());
		put.setDurability(Durability.SKIP_WAL);
		//				put.setWriteToWAL(false);
		//				put.addColumn("cf".getBytes(), "DOMAIN".getBytes(), splits[0].getBytes());
		//				put.addColumn("cf".getBytes(), "REPORTTIME".getBytes(), splits[1].getBytes());
		//				put.addColumn("cf".getBytes(), "IP".getBytes(), splits[2].getBytes());
		//				put.addColumn("cf".getBytes(), "HTTPNUM".getBytes(), num.getBytes());
		put.add("cf".getBytes(), "ATT".getBytes(), "cheng".getBytes());
		dao.save(put, "SDS_HTTP_STAT_H");
		//		dao.insert("test", "testrow", "cf", "c1", "chengbin");
		//		dao.incrementValue("test", "testrow", "cf", "inc", 2);
//		dao.incrementValue("t_ses_h", "182.254.241.239_2015091714", "SESSION", "MIDDLE",3);
//		List<Result> listSingle = dao.getRows("test", "testrow", new String[]{"c1"});
//		if(listSingle != null) {
//			Result result = listSingle.get(0);
//			for(KeyValue keyValue : result.raw()) {
//				String attnumStr = new String(keyValue.getValue());
//				System.out.println("num=" + attnumStr);
//			}
//		}
//		int num = dao.getAssignColumnNum("SDS_ABN_SESS_STAT_D", "20160301", "20160315", "DOS");
//		System.out.println("num=" + num);
//		dao.incerment("test", "2", "cf", "ips", 22);
////		System.out.println(dao.getRows("test", "increment", new String[]{"ips"}));
//		List<String> list = dao.getDataByCon("SDS_ABN_SESS_STAT_D", "20160311", "20160311");
//		for(Iterator<String> iter = list.iterator();iter.hasNext();) {
//			System.out.println(iter.next());
//		}
	}
}
